package com.cti.commons.mqtt.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * 线程池配置
 */
@Configuration
@ConfigurationProperties(prefix = "thread.pool")
public class MqttThreadPoolConfig {

    private static final Logger log = LoggerFactory.getLogger(MqttThreadPoolConfig.class);

    private static final int QUEUE_SIZE = 1000;

    private static final int CORE_SIZE = Runtime.getRuntime().availableProcessors();

    private static final int KEEP_ALIVE_TIME = 5;

    private static final int AWAIT_TERMINATION_SECONDS = 60;

    private static final int CORE_POOL_SIZE = 32 * CORE_SIZE;

    private static final int MAX_POOL_SIZE = 64 * CORE_SIZE;

    @Bean("mqttConsumeExecutor")
    public ThreadPoolTaskExecutor mqttConsumeExecutor() {
        log.debug("获取到当前cpu个数 '{}' .", CORE_SIZE);
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setRejectedExecutionHandler(new MqttConsumeExecutorRejectPolicy());
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(AWAIT_TERMINATION_SECONDS);
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
        taskExecutor.setQueueCapacity(QUEUE_SIZE);
        taskExecutor
            .setThreadFactory(new ThreadFactoryBuilder()
                .setNameFormat("mqttConsumeExecutor-%d")
                .setPriority(Thread.NORM_PRIORITY)
                .setUncaughtExceptionHandler((t, e) -> {
                    log.warn("mqttConsumeExecutor thread '{}' threw exception.", t.getName(), e);
                })
                .build());
        return taskExecutor;
    }

    @Bean("mqttSenderExecutor")
    public ThreadPoolTaskExecutor mqttSenderExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setRejectedExecutionHandler(new MqttConsumeExecutorRejectPolicy());
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(AWAIT_TERMINATION_SECONDS);
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
        taskExecutor.setQueueCapacity(QUEUE_SIZE);
        taskExecutor
            .setThreadFactory(new ThreadFactoryBuilder()
                .setNameFormat("mqttSenderExecutor-%d")
                .setPriority(Thread.NORM_PRIORITY)
                .setUncaughtExceptionHandler((t, e) -> {
                    log.warn("mqttSenderExecutor thread '{}' threw exception.", t.getName(), e);
                })
                .build());
        return taskExecutor;
    }

}
